package com.jason.library.flow

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.measureTimeMillis


/**
 * @description: FlowMain
 *
 * @author ChenBaoyang
 * @date 8/4/21 13:48
 */


fun sample() : List<Int> = listOf(1,2,3)

fun simple(): Sequence<Int> = sequence {
    for (i in 1..3) {
        Thread.sleep(100) // 假装我们正在计算
        yield(i)
    }
}

fun main() {
    /*sample().forEach {
        println(it)
    }*/

    /*simple().forEach {
        println(it)
    }*/
    //main2()
    //main3()
    //main4()
    //main5()
    //main6()
    //main7()
    //main8()
    //main9()
    //main10()
    //main11()
    //main12()
    //main13()
    //main14()
    //main15()
    //main16()
    //main17()
    //main18()
    //main19()
    //main20()
    //main21()
    //main22()
    //main23()
    //main24()
    //main25()
    //main26()
    //main27()
    //main28()
    //main29()
    //main30()
    //main31()
    main32()
}

suspend fun simple2(): List<Int> {
    delay(1000) // 假装我们在这里做了一些异步的事情
    return listOf(1, 2, 3)
}

fun main2() = runBlocking<Unit> {
    simple().forEach { value -> println(value) }
}

fun simple3(): Flow<Int> = flow {
    for (i in 3..5) {
        delay(100)
        emit(i)  // 发送下一个值
    }
}


fun main3() = runBlocking {
    // 启动并发的协程以验证主线程并未阻塞
    launch {
        for (k in 1..3) {
            println("I'm not blocked $k")
            delay(100)
        }
    }
    // 收集这个流
    simple3().collect {
            value -> println(value)
    }
}

/**
 * Flow 是一种类似于序列的冷流 — 这段 flow 构建器中的代码直到流被收集的时候才运行
 */
fun main4() = runBlocking<Unit> {
    println("Calling simple function...")
    val flow = simple3()
    println("Calling collect...")
    flow.collect { value -> println(value) }
    println("Calling collect again...")
    flow.collect { value -> println(value) }
}

fun simple4(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100)
        println("Emitting $i")
        emit(i)
    }
}

/**
 * 当 withTimeoutOrNull 块中代码在运行的时候流是如何在超时的情况下取消并停止执行
 */
fun main5() = runBlocking<Unit> {
    withTimeoutOrNull(250) { // 在 250 毫秒后超时
        simple4().collect { value -> println(value) }
    }
    println("Done")
}

/**
 * Flow流构建器
 * <p>
 *     flowOf 构建器定义了一个发射固定值集的流。
        使用 .asFlow() 扩展函数，可以将各种集合与序列转换为流
 * </p>
 * @return Flow<Int>
 */
fun simple5() : Flow<Int> {
    val flow = flowOf(1..3)
    return (1..3).asFlow()
}

suspend fun performRequest(request: Int): String {
    delay(1000) // 模仿长时间运行的异步工作
    return "response $request"
}

fun performFilter(input: Int): Boolean {
    return input > 1
}

/**
 * 过渡流操作符
 */
fun main6() = runBlocking<Unit> {
    (1..3).asFlow() // 一个请求流
        .filter { input -> performFilter(input) }
        .map { request -> performRequest(request) }
        .collect { response -> println(response) }
}

/**
 * 转换操作符
 */
fun main7() = runBlocking {
    (1..3).asFlow() // 一个请求流
        .transform { request ->
            emit("Making request $request")
            emit(performRequest(request))
        }
        .collect { response -> println(response) }
}

fun numbers(): Flow<Int> = flow {
    try {
        emit(1)
        emit(2)
        println("This line will not execute")
        emit(3)
    } finally {
        println("Finally in numbers")
    }
}

/**
 * 限长操作符(take)
 */
fun main8() = runBlocking<Unit> {
    numbers()
        .take(2) // 只获取前两个
        .collect { value -> println(value) }
}

/**
 * 末端流操作符
 * 末端操作符是在流上用于启动流收集的挂起函数。 collect 是最基础的末端操作符，但是还有另外一些更方便使用的末端操作符：
 * 1.转化为各种集合，例如 toList 与 toSet。
 * 2.获取第一个（first）值与确保流发射单个（single）值的操作符。
 * 3.使用 reduce 与 fold 将流规约到单个值。
 */
fun main9() = runBlocking {
    val sum = (1..5).asFlow()
        .map { it * it } // 数字 1 至 5 的平方
        .reduce { a, b -> a + b } // 求和（末端操作符）
    println(sum)

    val list = (1..5).asFlow()
        .map { it * it } // 数字 1 至 5 的平方
        .toList()
    println(list)
}

/**
 * 流的每次单独收集都是按顺序执行的
 */
fun main10() = runBlocking {

    (1..5).asFlow()
        .filter {
            println("Filter $it")
            it % 2 == 0
        }
        .map {
            println("Map $it")
            "string $it"
        }.collect {
            println("Collect $it")
        }
}

/**
 * flowOn函数用于更改流发射的上下文（类似rxJava切换线程）
 */
fun simple11(): Flow<Int> = flow {
    for (i in 1..5) {
        delay(100)
        emit(i)
    }
}.flowOn(Dispatchers.Default)

fun main11() = runBlocking {
    simple11().collect {
        println(it)
    }
}

fun simple12(): Flow<Int> = flow {
    for (i in 1..5) {
        delay(100) // 假装我们异步等待了 100 毫秒
        emit(i) // 发射下一个值
    }
}

/**
 * buffer函数用来缓冲发射项
 */
fun main12() = runBlocking<Unit> {
    val time = measureTimeMillis {
        simple12()
            .buffer() // 缓冲发射项，无需等待
            .collect { value ->
            delay(300) // 假装我们花费 300 毫秒来处理它
            println(value)
        }
    }
    println("Collected in $time ms")
}

/**
 * conflate合并函数：合并发射项 - 加快处理速度，
 * 合并是加快处理速度的一种方式。它通过删除发射值来实现
 *
 */
fun main13() = runBlocking {
    val time = measureTimeMillis {
        simple12()
            .conflate() // 合并发射项，不对每个值进行处理，
            .collect { value ->
                delay(300) // 假装我们花费 300 毫秒来处理它
                println(value)
            }
    }
    println("Collected in $time ms")
}

/**
 * collectLatest函数-处理最新值
 * 这种方式是取消缓慢的收集器，并在每次发射新值的时候重新启动它
 *
 */
fun main14() = runBlocking {
    val time = measureTimeMillis {
        simple12()
            .collectLatest { value -> // 取消并重新发射最后一个值
                println("Collecting $value")
                delay(300) // 假装我们花费 300 毫秒来处理它
                println("Done $value")
            }
    }
    println("Collected in $time ms")
}

/**
 * zip函数：用于组合两个流中的相关值
 */
fun main15() = runBlocking {
    val num = (1..3).asFlow()
    val str = flowOf("a","b","c")
    num.zip(str){ a,b -> "$a -> $b" }.collect {
        println(it)
    }
}

/**
 * Combine函数：当流表示一个变量或操作的最新值时，
 * 可能需要执行计算，这依赖于相应流的最新值，并且每当上游流产生值的时候都需要重新计算
 */
fun main16() = runBlocking {

    val nums = (1..3).asFlow().onEach { delay(300) } // 发射数字 1..3，间隔 300 毫秒
    val strs = flowOf("one", "two", "three").onEach { delay(400) } // 每 400 毫秒发射一次字符串
    //未使用Combine函数更新计算
    val startTime = System.currentTimeMillis() // 记录开始的时间
    nums.zip(strs) { a, b -> "$a -> $b" } // 使用“zip”组合单个字符串
        .collect { value -> // 收集并打印
            println("$value at ${System.currentTimeMillis() - startTime} ms from start")
        }

    println("----------------分界线------------------")
    //使用Combine的
    val startTime2 = System.currentTimeMillis() // 记录开始的时间
    nums.combine(strs) { a, b -> "$a -> $b" } // 使用“combine”组合单个字符串
        .collect { value -> // 收集并打印
            println("$value at ${System.currentTimeMillis() - startTime2} ms from start")
        }
}

/*********************************平展流**********************************/

fun requestFlow(i: Int): Flow<String> = flow {
    emit("$i: First")
    delay(500) // 等待 500 毫秒
    emit("$i: Second")
}

/**
 * 平展流
 */
fun main17() = runBlocking {
    (1..3).asFlow().map { requestFlow(it) }
}

/**
 * flatMapConcat函数:平展开
 */
fun main18() = runBlocking {
    val startTime = System.currentTimeMillis() // 记录开始时间
    (1..3).asFlow().onEach { delay(100) } // 每 100 毫秒发射一个数字
        .flatMapConcat { requestFlow(it) }
        .collect { value -> // 收集并打印
            println("$value at ${System.currentTimeMillis() - startTime} ms from start")
        }
}

/**
 * flatMapMerge函数：另一种展平模式是并发收集所有传入的流，并将它们的值合并到一个单独的流，以便尽快的发射值
 */
fun main19() = runBlocking {
    val startTime = System.currentTimeMillis() // 记录开始时间
    (1..3).asFlow().onEach { delay(100) } // 每 100 毫秒发射一个数字
        .flatMapMerge { requestFlow(it) }
        .collect { value -> // 收集并打印
            println("$value at ${System.currentTimeMillis() - startTime} ms from start")
        }
}

/**
 * flatMapLatest
 * 与 collectLatest 操作符类似，也有相对应的“最新”展平模式，在`发出新流后立即取消先前流`的收集
 */
fun main20() = runBlocking {
    val startTime  = System.currentTimeMillis()
    (1..3).asFlow().onEach { delay(100) }
        .flatMapLatest { requestFlow(it) }
        .collect {
                value -> // 收集并打印
            println("$value at ${System.currentTimeMillis() - startTime} ms from start")
        }
}


/*************************流异常***************************/

fun simple21(): Flow<Int> = flow {
    for (i in 1..3) {
        println("Emitting $i")
        emit(i) // 发射下一个值
    }
}
/**
 * 收集器 try 与 catch
 */
fun main21() = runBlocking {
    try {
        simple21().collect { value ->
            println(value)
            //不满足条件抛出异常
            check(value <= 1) { "Collected $value" }
        }
    } catch (e: Throwable) {
        println("Caught $e")
    }
}

fun simple22(): Flow<String> =
    flow {
        for (i in 1..3) {
            println("Emitting $i")
            emit(i) // 发射下一个值
        }
    }
        .map { value ->
            check(value <= 1) { "Crashed on $value" }
            "string $value"
        }

/**
 * 一切都已捕获
 */
fun main22() = runBlocking {
    try {
        simple22().collect { value -> println(value) }
    } catch (e: Throwable) {
        println("Caught $e")
    }
}

/**
 * 发射一个异常
 * @return [ERROR : Error function type]
 */
fun main23() = runBlocking {
    simple22()
        .catch { e -> emit("Caught $e") } // 发射一个异常
        .collect { value -> println(value) }
}

fun simple24(): Flow<Int> = flow {
    for (i in 1..3) {
        println("Emitting $i")
        emit(i)
    }
}

/**
 * 透明捕获
 * catch 过渡操作符遵循异常透明性，仅捕获上游异常（catch 操作符上游的异常，但是它下面的不是）。
 * 如果 collect { ... } 块（位于 catch 之下）抛出一个异常，那么异常会逃逸
 */
fun main24() = runBlocking {
    simple24()
        .catch { e -> println("Caught $e") } // 不会捕获下游异常
        .collect { value ->
            check(value <= 1) { "Collected $value" }
            println(value)
        }
}

/**
 * 声明式捕获
 * 将 catch 操作符的声明性与处理所有异常的期望相结合，将 collect 操作符的代码块移动到 onEach 中，
 * 并将其放到 catch 操作符之前。收集该流必须由调用无参的 collect() 来触发
 */
fun main25() = runBlocking {
    simple24()
        .onEach { value ->
            check(value <= 1) { "Collected $value" }
            println(value)
        }
        .catch { e -> println("Caught $e") }
        .collect()
}

/***************************流完成******************************/
/**当流收集完成时（普通情况或异常情况），它可能需要执行一个动作*/

/**
 * 命令式 finally 块
 */

fun simple26(): Flow<Int> = (1..3).asFlow()

fun main26() = runBlocking {
    try {
        simple26().collect { value -> println(value) }
    } finally {
        println("Done")
    }
}

/**
 * 声明式处理
 */
fun main27() = runBlocking {
    simple26()
        .onCompletion { println("Done") }
        .collect { value -> println(value) }
}


fun simple28(): Flow<Int> = flow {
    emit(1)
    throw RuntimeException()
}

/**
 * onCompletion 的主要优点是其 lambda 表达式的可空参数 Throwable 可以用于确定流收集是正常完成还是有异常发生
 */
fun main28() = runBlocking {
    simple28()
        .onCompletion { cause -> if (cause != null) println("Flow completed exceptionally") }
        .catch { cause -> println("Caught exception") }
        .collect { value -> println(value) }
}

/**
 * 与 catch 操作符的另一个不同点是 onCompletion 能观察到所有异常并且仅在上游流成功完成（没有取消或失败）的情况下接收一个 null 异常
 */
fun main29() = runBlocking {
    (1..3).asFlow()
        .onCompletion { cause -> println("Flow completed with $cause") }
        .collect { value ->
            check(value <= 1) { "Collected $value" }
            println(value)
        }
}

// 模仿事件流
fun events(): Flow<Int> = (1..3).asFlow().onEach { delay(100) }

/**
 * 启动流：collect和launchIn
 * 1.使用 collect 末端操作符，那么后面的代码会一直等待直至流被收集
 * 2.使用 launchIn 替换 collect 我们可以在单独的协程中启动流的收集，这样就可以立即继续进一步执行代码
 */
fun main30() = runBlocking {
    events()
        .onEach { event -> println("Event: $event") }
        .collect() // <--- 等待流收集
    println("Done")

    println("——————————————————————————————————————————")

    events()
        .onEach { event -> println("Event: $event") }
        .launchIn(this) // <--- 等待流收集
    println("Done")
}

fun foo(): Flow<Int> = flow {
    for (i in 1..5) {
        println("Emitting $i")
        emit(i)
    }
}

/**
 * 流取消：常规的取消是无法取消的，会报错
 */
fun main31() = runBlocking<Unit> {
    foo().collect { value ->
        if (value == 3) cancel()
        println(value)
    }
}


fun main32() = runBlocking<Unit> {
    (1..5).asFlow().cancellable().collect { value ->
        if (value == 3) cancel()
        println(value)
    }
}
